//! 安全的事件分发器实现
//! 
//! 使用标准库的MPSC通道，避免自定义无锁队列的内存安全问题

use crate::event::{DynamicEvent, DynamicEventHandler};

use std::sync::{Arc, RwLock};
use fxhash::FxHashMap;
use std::thread;
use std::time::Duration;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};

/// 安全分发器配置
#[derive(Debug, Clone)]
pub struct SafeDispatcherConfig {
    pub worker_count: usize,
    pub channel_capacity: usize,
}

impl Default for SafeDispatcherConfig {
    fn default() -> Self {
        Self {
            worker_count: 2,
            channel_capacity: 10000,
        }
    }
}

/// 安全的事件分发器（使用标准库MPSC）
pub struct SafeEventDispatcher {
    config: SafeDispatcherConfig,
    sender: Sender<DynamicEvent>,
    handlers: Arc<RwLock<FxHashMap<String, Vec<Arc<dyn DynamicEventHandler>>>>>,
    running: Arc<AtomicBool>,
    worker_handles: Vec<thread::JoinHandle<()>>,
}

/// 安全的发送器
#[derive(Clone)]
pub struct SafeEventSender {
    sender: Sender<DynamicEvent>,
}

impl SafeEventSender {
    pub fn publish(&self, event: DynamicEvent) -> Result<(), DynamicEvent> {
        self.sender.send(event).map_err(|e| e.0)
    }
}

impl SafeEventDispatcher {
    pub fn new(config: SafeDispatcherConfig) -> Self {
        let (sender, _receiver) = mpsc::sync_channel(config.channel_capacity);
        
        Self {
            config,
            sender,
            handlers: Arc::new(RwLock::new(FxHashMap::default())),
            running: Arc::new(AtomicBool::new(false)),
            worker_handles: Vec::new(),
        }
    }

    pub fn register_handler(&mut self, event_type: &str, handler: Arc<dyn DynamicEventHandler>) {
        let mut handlers = self.handlers.write().unwrap();
        handlers.entry(event_type.to_string())
            .or_insert_with(Vec::new)
            .push(handler);
    }

    pub fn clone_sender(&self) -> SafeEventSender {
        SafeEventSender {
            sender: self.sender.clone(),
        }
    }

    pub fn start(&mut self) {
        if self.running.load(Ordering::Relaxed) {
            return;
        }

        self.running.store(true, Ordering::Relaxed);

        // 为每个worker创建独立的receiver
        let mut receivers = Vec::new();
        for _ in 0..self.config.worker_count {
            let (sender, receiver) = mpsc::sync_channel(self.config.channel_capacity);
            receivers.push((sender, receiver));
        }

        // 分发器线程：负责将事件分发给各个worker
        let main_receiver = self.sender.clone();
        let worker_senders: Vec<_> = receivers.iter().map(|(s, _)| s.clone()).collect();
        let running_clone = self.running.clone();
        
        thread::Builder::new()
            .name("safe-dispatcher".to_string())
            .spawn(move || {
                let mut worker_index = 0;
                while running_clone.load(Ordering::Relaxed) {
                    // 注意：这里需要一个不同的接收方式
                    // 由于我们使用的是sender，需要重新设计
                    thread::sleep(Duration::from_millis(1));
                }
            })
            .expect("Failed to spawn dispatcher thread");

        // Worker线程
        for (worker_id, (_sender, receiver)) in receivers.into_iter().enumerate() {
            let running = self.running.clone();
            let handlers = self.handlers.clone();

            let handle = thread::Builder::new()
                .name(format!("safe-worker-{}", worker_id))
                .spawn(move || {
                    while running.load(Ordering::Relaxed) {
                        match receiver.try_recv() {
                            Ok(event) => {
                                let event_type = event.event_type();
                                let handlers_guard = handlers.read().unwrap();
                                
                                if let Some(handler_list) = handlers_guard.get(event_type) {
                                    for handler in handler_list {
                                        let _result = handler.handle_dynamic(&event);
                                        break; // 只处理第一个匹配的handler
                                    }
                                }
                            }
                            Err(mpsc::TryRecvError::Empty) => {
                                thread::sleep(Duration::from_millis(1));
                            }
                            Err(mpsc::TryRecvError::Disconnected) => break,
                        }
                    }
                })
                .expect("Failed to spawn worker thread");

            self.worker_handles.push(handle);
        }
    }

    pub fn stop(&mut self) {
        self.running.store(false, Ordering::Relaxed);
        
        while let Some(handle) = self.worker_handles.pop() {
            let _ = handle.join();
        }
    }

    pub fn publish(&self, event: DynamicEvent) -> Result<(), DynamicEvent> {
        self.sender.send(event).map_err(|e| e.0)
    }
}